package com.haibara.codesubmiteval.kafka.processor;

import cn.hutool.json.JSONUtil;
import com.haibara.codesubmiteval.constant.KafkaConstant;
import com.haibara.codesubmiteval.judge.model.ExecuteCodeRequest;
import com.haibara.codesubmiteval.judge.model.ExecuteCodeRespond;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * 用于发送消息至Kafka的类
 *
 * @Author: mengyu
 * @Date: 2025/2/10
 */
@Component
@Slf4j
public class Processor {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 发送消息
     *
     * @param topic 主题
     * @param msg   信息
     */
    public void sendMsg(String topic, String msg) {
        CompletableFuture<SendResult<String, String>> send = this.kafkaTemplate.send(topic, msg);
        send.thenAcceptAsync(result -> {
            log.info(topic + "发送成功!");
        }).exceptionally(ex -> {
            log.error(topic + " 发送失败!");
            ex.printStackTrace();
            return null;
        });
    }

    // SubmitTopic
    public void sendMsg(Long questionSubmitId) {
        CompletableFuture<SendResult<String, String>> send = this.kafkaTemplate.send(KafkaConstant.SUBMIT_TOPIC, String.valueOf(questionSubmitId));
        send.thenAcceptAsync(result -> {
            log.info("topic:{} ,id: {} ,发送成功!", KafkaConstant.SUBMIT_TOPIC, questionSubmitId);
        }).exceptionally(ex -> {
            log.error("topic:{} ,id: {} ,发送失败!", KafkaConstant.SUBMIT_TOPIC, questionSubmitId);
            ex.printStackTrace();
            return null;
        });
    }

    // RunCodeTopic
    public void sendMsg(ExecuteCodeRequest ecr) {
        CompletableFuture<SendResult<String, String>> send = this.kafkaTemplate.send(KafkaConstant.RUN_CODE_TOPIC, JSONUtil.toJsonStr(ecr));
        send.thenAcceptAsync(result -> {
            log.info("topic:{} ,id: {} ,发送成功!", KafkaConstant.RUN_CODE_TOPIC, ecr.getQuestionSubmitId());
        }).exceptionally(ex -> {
            log.error("topic:{} ,id: {} ,发送失败!", KafkaConstant.RUN_CODE_TOPIC, ecr.getQuestionSubmitId());
            ex.printStackTrace();
            return null;
        });
    }

    // JudgeTopic
    public void sendMsg(ExecuteCodeRespond ecr) {
        CompletableFuture<SendResult<String, String>> send = this.kafkaTemplate.send(KafkaConstant.JUDGE_TOPIC, JSONUtil.toJsonStr(ecr));
        send.thenAcceptAsync(result -> {
            log.info("topic:{} ,id: {} ,发送成功!", KafkaConstant.JUDGE_TOPIC, ecr.getQuestionSubmitId());
        }).exceptionally(ex -> {
            log.error("topic:{} ,id: {} ,发送失败!", KafkaConstant.JUDGE_TOPIC, ecr.getQuestionSubmitId());
            ex.printStackTrace();
            return null;
        });
    }


    /**
     * 将消息发送到死信队列
     *
     * @param msg 消息
     */
    public void sendMsgToDLQ(String msg) {
        kafkaTemplate.send(KafkaConstant.DEAD_LETTER, msg);
    }

    public void sendMsgToDLQ(Object object) {
        kafkaTemplate.send(KafkaConstant.DEAD_LETTER, JSONUtil.toJsonStr(object));
    }
}
